// @Author EthanScriptOn
// @Desc
package connect

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/switch_log"
	"gitee.com/fatzeng/srf_switch_basic_components/util"
	"gitee.com/fatzeng/srf_switch_expansion_package/detect"
	"github.com/streadway/amqp"
)

var RabbitmqProducer *util.OnceInitializer

var RabbitmqProducerChecker = func() (isInvalid bool) {
	producer, err := generateRabbitmqProducer()
	if err != nil {
		return true
	}
	if !isChannelOpen(producer) {
		_ = producer.Close()
		return true
	}
	return false
}

var RabbitmqProducerRecover = func() (err error) {
	RabbitmqProducer.Reset()
	_, err = RabbitmqProducer.Get()
	return err
}
var RabbitmqProducerMonitor = detect.GenerateResourceMonitor("rabbitmq_producer", true, RabbitmqProducerChecker, RabbitmqProducerRecover, detect.MonitorFunctionInstance)

func GetRabbitmqProducer() (instance *amqp.Channel, err error) {
	defer detect.BackendHealthChecker.AddResource(RabbitmqProducerMonitor)
	return generateRabbitmqProducer()
}

func isChannelOpen(ch *amqp.Channel) (isOpen bool) {
	err := ch.Ack(uint64(0), false)
	if err != nil {
		if amqpErr, ok := err.(*amqp.Error); ok {
			if amqpErr == amqp.ErrClosed {
				return false
			}
			switch_log.Logger().Error("AMQP error: %v", amqpErr)
			return false
		} else {
			switch_log.Logger().Error("Unexpected error: %v", err)
			return false
		}
	}
	return true
}

func generateRabbitmqProducer() (producer *amqp.Channel, err error) {
	if RabbitmqProducer == nil {
		return nil, fmt.Errorf("the RabbitmqProducer initialization required")
	}
	instanceInter, err := RabbitmqProducer.Get()
	if err != nil {
		return nil, err
	}
	instance, ok := instanceInter.(*amqp.Channel)
	if !ok {
		return nil, fmt.Errorf("the instance is not of type *amqp.Channel")
	}
	return instance, nil
}

var RabbitmqConsumer *util.OnceInitializer

var RabbitmqConsumerChecker = func() (isInvalid bool) {
	consumer, err := generateRabbitmqConsumer()
	if err != nil {
		return true
	}
	if !isChannelOpen(consumer) {
		_ = consumer.Close()
		return true
	}
	return false
}

var RabbitmqConsumerRecover = func() (err error) {
	RabbitmqConsumer.Reset()
	_, err = RabbitmqConsumer.Get()
	return err
}

var RabbitmqConsumerMonitor = detect.GenerateResourceMonitor("rabbitmq_consumer", true, RabbitmqConsumerChecker, RabbitmqConsumerRecover, detect.MonitorFunctionInstance)

func GetRabbitmqConsumer() (instance *amqp.Channel, err error) {
	defer detect.BackendHealthChecker.AddResource(RabbitmqConsumerMonitor)
	return generateRabbitmqProducer()
}

func generateRabbitmqConsumer() (producer *amqp.Channel, err error) {
	if RabbitmqConsumer == nil {
		return nil, fmt.Errorf("the RabbitmqConsumer initialization required")
	}
	instanceInter, err := RabbitmqConsumer.Get()
	if err != nil {
		return nil, err
	}
	instance, ok := instanceInter.(*amqp.Channel)
	if !ok {
		return nil, fmt.Errorf("the instance is not of type *amqp.Channel")
	}
	return instance, nil
}
